技术|使用FlinkSQL传输市场数据1：传输VWAP

这些列是：交易品种，价格，数量，出价 ID，要价 ID，买方 ID，卖方 ID，步骤和时间戳。步骤列是离散步骤 ABM 市场模拟的伪像，出于我们的目的可以忽略；其余各栏不言自明。

要处理此数据，我们需要通过发出 CREATE TABLE 语句来声明 Flink SQL 表。我们的示例数据是基于文件系统的，但是可以轻松更改连接器类型以从其他来源（例如Kafka主题）读取数据。请注意，event_time 是派生的列，也用于水印。通过加水印，Flink 可以限制等待延迟到达和故障事件的时间，以便可以取得进展。在这里，我们声明，到达 event_time 超过水印一分钟以上的记录将被忽略。

CREATE TABLE trades (
symbol STRING,
price DOUBLE,
vol INT,
bid_id STRING,
ask_id STRING,
buyer_id STRING,
seller_id STRING,
step INT,
ts_str STRING,
event_time AS TO_TIMESTAMP (ts_str, 'dd-MMM-yyyy HH:mm:ss.SSS'),
WATERMARK FOR event_time AS event_time - INTERVAL '1' MINUTE
) WITH (
'connector' = 'filesystem',
'path' = '/path/to/varstream/data/trades_raw',
'format' = 'csv'
);

VWAP 的公式很简单：对于指定时间段内的每笔交易，将价格乘以交易股份数即可。将其总和除以该时间段内已交易的股票总数。下面的流查询将显示当前的 VWAP，它将随着新交易事件的到来而更新：

SELECT
symbol,
SUM (vol)                     AS cumulative_volume,
SUM (price * vol)             AS cumulative_pv,
SUM (price * vol) / SUM (vol) AS vwap
FROM
trades
GROUP BY
symbol
;

您可以通过发出一个简单的查询来验证事件的重播方式：

SELECT * FROM trades_replay

使用此视图，我们现在可以发出相同的 VWAP 聚合查询，并观察对 VWAP 的流更新，就好像它们是实时发生的一样：

SELECT
symbol,
SUM (vol) AS cumulative_volume,
SUM (price * vol) AS cumulative_pv,
SUM (price * vol) SUM (vol) AS vwap
FROM
trades_replay
GROUP BY
symbol
;

前面的示例显示了如何计算当天的流式 VWAP。假设您要以每隔1分钟的时间建立一个带有蜡烛图的交易仪表板。您可能需要计算每分钟的 VWAP、高价、低价和总体积。Flink SQL 通过组窗口使此操作变得容易，组窗口可以在 GROUP BY 时间间隔上应用聚合函数。

下面显示了如何获取每分钟的 VWAP：

CREATE VIEW vwap_1m AS (
SELECT
symbol,
TUMBLE_START (event_time, INTERVAL '1' MINUTES) AS start_time,
TUMBLE_ROWTIME (event_time, INTERVAL '1' MINUTES) AS row_time,
MAX (price) AS max_price,
MIN (price) AS min_price,
SUM (price * vol) AS total_price,
SUM (vol) AS total_vol,
SUM (price * vol) SUM (vol) AS vwap
FROM
trades
GROUP BY
TUMBLE (event_time, INTERVAL '1' MINUTES), symbol
);
SELECT symbol, start_time, total_price, total_vol, vwap FROM vwap_1m ;

前面的操作为每分钟内发生的交易计算了 VWAP。如果要在几分钟内计算移动的 VWAP（MVWAP），则 Flink SQL 提供了一个跳跃的组窗口。下面显示了5分钟的移动 VWAP，步长为1分钟。